package lucene.server.web.task;

import lombok.extern.slf4j.Slf4j;
import lucene.server.common.event.EventData;
import lucene.server.common.queue.ChangeProcessor;
import lucene.server.common.queue.FileQueue;
import lucene.server.common.task.ITask;
import lucene.server.common.utils.ThreadUtil;
import lucene.server.core.LuceneIndex;
import lucene.server.core.LuceneService;
import lucene.server.core.model.IndexConfig;
import java.util.ArrayList;
import java.util.List;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import java.util.concurrent.ScheduledFuture;

/**
 *
 * */
@Slf4j
public class DataChangeTask implements Runnable , ITask {

    private int batchSize = 6000;
    private FileQueue<EventData> fileQueue;
    private boolean isShutdown = false;
    private ScheduledFuture future;
    private boolean running = false;
    private IndexConfig indexConfig;
    public DataChangeTask(IndexConfig indexConfig) {
        this.indexConfig = indexConfig;
        fileQueue = ChangeProcessor.getFileQueue(indexConfig.getDatabase(),indexConfig.getTable());
    }

    @Override
    public void run() {
        if (running){
            return;
        }
        if (fileQueue.isEmpty()) {
            //如果系统准备关闭则直接返回
            if (!isShutdown){ ThreadUtil.sleep(1000);}
            return;
        }
        running = true;
        if (isShutdown){ log.info("cdc 处理后续数据");}
        List<EventData> insertList = new ArrayList<>();
        List<EventData> updateList = new ArrayList<>();
        List<EventData> deleteList = new ArrayList<>();
        for (int i = 0; i < batchSize; i++){
            EventData eventData = fileQueue.poll();
            if (eventData == null){
                break;
            }

            switch (eventData.getTaskType()){
                case INSERT:
                    insertList.add(eventData);
                    break;
                case UPDATE:
                    updateList.add(eventData);
                    break;
                case DELETE:
                    deleteList.add(eventData);
                    break;
            }
        }
        LuceneIndex basicService = LuceneService.getIndex(this.indexConfig.getName());
        try {
            basicService.getInsertService().addEventData(insertList);
        }
        catch (Exception e) {
            log.error("cdc {} insert error",getName(),e);
        }
        try {
            basicService.getUpdateService().addEventData(updateList);
        } catch (Exception e) {
            log.error("cdc {} update error",getName(),e);
        }
        try {
            basicService.getDeleteService().addEventData(deleteList);
        } catch (Exception e) {
            log.error("cdc {} delete error",getName(),e);
        }
        insertList.clear();updateList.clear();deleteList.clear();
        running = false;
    }

    @Override
    public String getName() {
        return "change_"+indexConfig.getName();
    }

    @Override
    public ScheduledFuture createScheduled(ThreadPoolTaskScheduler threadPoolTaskScheduler) {
        this.future = threadPoolTaskScheduler.scheduleWithFixedDelay(this, 500);
        return this.future;
    }

    @Override
    public String getTaskPlan() {
        return "Delay:500毫秒";
    }

    @Override
    public void shutdown() {
        try {
            isShutdown = true;
            log.info("cdc task {} shutdown",indexConfig.getName());
            run();
            if (this.future != null) {
                this.future.cancel(true);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    @Override
    public boolean isRunning() {
        return running;
    }
}